package com.yejiarong.weather.agent.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.nacos.common.utils.ThreadFactoryBuilder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.yejiarong.common.constant.ElasticConstants;
import com.yejiarong.common.exception.ResponseCode;
import com.yejiarong.common.tools.HttpResult;
import com.yejiarong.elasticsearch.api.bean.request.DeleteByQueryForm;
import com.yejiarong.elasticsearch.api.bean.request.SaveRequestForm;
import com.yejiarong.elasticsearch.api.bean.request.query.Query;
import com.yejiarong.elasticsearch.api.bean.response.DeleteByQueryResult;
import com.yejiarong.elasticsearch.api.enums.QueryType;
import com.yejiarong.elasticsearch.api.service.ElasticService;
import com.yejiarong.influxdb.api.bean.request.SavePointRequest;
import com.yejiarong.influxdb.api.service.InfluxdbService;
import com.yejiarong.weather.agent.common.bridge.IWeatherPythonService;
import com.yejiarong.weather.agent.bean.dto.CountyWeatherDto;
import com.yejiarong.weather.agent.bean.entity.WeatherRegionCode;
import com.yejiarong.weather.agent.mapper.IWeatherRegionCodeMapper;
import com.yejiarong.weather.agent.service.IWeatherDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author YeJR
 */
@Slf4j
@Service
public class WeatherDataServiceImpl implements IWeatherDataService {

    private final ThreadFactory threadFactory = new ThreadFactoryBuilder().nameFormat("sync-weather-data-%d").build();

    private final ThreadPoolExecutor threadPoolExecutor =
            new ThreadPoolExecutor(2, 2, 60,
                    TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10000), threadFactory);

    private final IWeatherRegionCodeMapper weatherRegionCodeMapper;

    private final IWeatherPythonService weatherPythonService;

    private final ElasticService elasticService;

    private final InfluxdbService influxdbService;

    public WeatherDataServiceImpl(IWeatherRegionCodeMapper weatherRegionCodeMapper,
                                  IWeatherPythonService weatherPythonService, ElasticService elasticService, InfluxdbService influxdbService) {
        this.weatherRegionCodeMapper = weatherRegionCodeMapper;
        this.weatherPythonService = weatherPythonService;
        this.elasticService = elasticService;
        this.influxdbService = influxdbService;
    }

    /**
     * 收集天气数据
     */
    @Override
    public void gatherWeatherData() {
        // 遍历现有的区域code
        List<WeatherRegionCode> weatherRegionCodes = weatherRegionCodeMapper.selectList(Wrappers.emptyWrapper());
        for (WeatherRegionCode weatherRegionCode : weatherRegionCodes) {
            threadPoolExecutor.submit(() -> gatherWeatherData(weatherRegionCode));
        }
    }

    /**
     * 收集某个城市的天气
     *
     * @param city 城市
     */
    @Override
    public void gatherWeatherData(String city) {
        LambdaQueryWrapper<WeatherRegionCode> lambdaQueryWrapper = new LambdaQueryWrapper<WeatherRegionCode>().eq(WeatherRegionCode::getCity, city);
        List<WeatherRegionCode> weatherRegionCodes = weatherRegionCodeMapper.selectList(lambdaQueryWrapper);
        for (WeatherRegionCode weatherRegionCode : weatherRegionCodes) {
            threadPoolExecutor.submit(() -> gatherWeatherData(weatherRegionCode));
        }
    }

    /**
     * 清理记录的未来天气
     */
    @Override
    public void cleanOutdatedFutureWeatherData() {
        long limitTime = LocalDate.now().plusDays(-1).atStartOfDay().toInstant(ZoneOffset.of("+8")).toEpochMilli();

        List<Object> values = new ArrayList<>();
        values.add(null);
        values.add(limitTime);

        Query query = new Query();
        query.setType(QueryType.NUMBER_RANGE);
        query.setField("updateTime");
        query.setValues(values);

        // 删除日数据
        DeleteByQueryForm deleteByQueryForm = new DeleteByQueryForm();
        deleteByQueryForm.setIndex(ElasticConstants.WEATHER_DAY_DATA);
        deleteByQueryForm.setQueries(List.of(query));

        HttpResult<DeleteByQueryResult> deleteByQueryResultHttpResult1 = elasticService.deleteDocumentByQuery(deleteByQueryForm);
        if (Objects.nonNull(deleteByQueryResultHttpResult1) && deleteByQueryResultHttpResult1.isSuccess() && Objects.nonNull(deleteByQueryResultHttpResult1.getData())) {
            DeleteByQueryResult deleteByQueryResult = deleteByQueryResultHttpResult1.getData();
            log.info("{}总计：{}条数据，删除{}条数据", ElasticConstants.WEATHER_DAY_DATA, deleteByQueryResult.getTotal(), deleteByQueryResult.getDeleted());
        }

        // 删除小时级别数据
        deleteByQueryForm.setIndex(ElasticConstants.WEATHER_HOUR_DATA);
        HttpResult<DeleteByQueryResult> deleteByQueryResultHttpResult2 = elasticService.deleteDocumentByQuery(deleteByQueryForm);
        if (Objects.nonNull(deleteByQueryResultHttpResult2) && deleteByQueryResultHttpResult2.isSuccess() && Objects.nonNull(deleteByQueryResultHttpResult2.getData())) {
            DeleteByQueryResult deleteByQueryResult = deleteByQueryResultHttpResult2.getData();
            log.info("{}总计：{}条数据，删除{}条数据", ElasticConstants.WEATHER_HOUR_DATA, deleteByQueryResult.getTotal(), deleteByQueryResult.getDeleted());
        }
    }


    /**
     * 收集天气信息
     * @param weatherRegionCode weatherRegionCode
     */
    private void gatherWeatherData(WeatherRegionCode weatherRegionCode) {
        try {
            Map<String, String> params = new HashMap<>();
            params.put("code", weatherRegionCode.getCode());
            CountyWeatherDto countyWeathers = weatherPythonService.getCountyWeather(params);
            // 保存未来7天的天气数据
            saveFuture7dData(weatherRegionCode, countyWeathers.getFuture7dData());
            // 保存未来整点天气数据
            saveFutureHourData(weatherRegionCode, countyWeathers.getFutureHourData());
            // 保存过去的天气数据
            saveLast24hData(weatherRegionCode, countyWeathers.getLast24hData());
            log.info("{}-{}-{}的天气同步完成", weatherRegionCode.getProvince(), weatherRegionCode.getCity(), weatherRegionCode.getCounty());
            // 随机等待，防止访问太频繁
            Thread.sleep((long) (Math.random() * 1000));
        } catch (Exception e) {
            log.error("收集天气数据异常", e);
        }
    }

    /**
     * 保存过去24小时的天气数据
     * @param weatherRegionCode 区域code
     * @param last24hData 天气数据
     */
    private void saveLast24hData(WeatherRegionCode weatherRegionCode, List<CountyWeatherDto.LastHourWeather> last24hData) {
        if (CollectionUtils.isEmpty(last24hData)) {
            return;
        }

        for (CountyWeatherDto.LastHourWeather lastHourWeather : last24hData) {
            SavePointRequest savePointRequest = new SavePointRequest();
            savePointRequest.setMetric("weather");
            savePointRequest.setTime(lastHourWeather.getTime());
            savePointRequest.addTag("province", weatherRegionCode.getProvince())
                    .addTag("city", weatherRegionCode.getCity()).addTag("county", weatherRegionCode.getCounty())
                    .addField("temperature", lastHourWeather.getTemperature()).addField("humidity", lastHourWeather.getHumidity())
                    .addField("windDirection", lastHourWeather.getWindDirection()).addField("windSpeed", lastHourWeather.getWindSpeed());
            influxdbService.collectPoint("weather", savePointRequest);
        }

    }

    /**
     * 保存未来小时数据
     * @param weatherRegionCode 区域code
     * @param futureHourData 天气数据
     */
    private void saveFutureHourData(WeatherRegionCode weatherRegionCode, List<CountyWeatherDto.FutureHourWeather> futureHourData) {
        if (CollectionUtils.isEmpty(futureHourData)) {
            return;
        }

        for (CountyWeatherDto.FutureHourWeather data : futureHourData) {
            String docId = weatherRegionCode.getProvince() + ":" + weatherRegionCode.getCity() + ":"
                    + weatherRegionCode.getCounty() + ":" + data.getTime().substring(0, 16);

            JSONObject jsonObject = JSONObject.from(data);
            jsonObject.put("time", data.getTime().substring(0, 16));
            jsonObject.put("province", weatherRegionCode.getProvince());
            jsonObject.put("city", weatherRegionCode.getCity());
            jsonObject.put("county", weatherRegionCode.getCounty());
            jsonObject.put("updateTime", new Date().getTime());

            HttpResult<Boolean> existsResult = elasticService.documentExists(ElasticConstants.WEATHER_HOUR_DATA, docId);
            if (Objects.nonNull(existsResult) && existsResult.getCode() == ResponseCode.SUCCESS.getCode() && Objects.nonNull(existsResult.getData())) {
                SaveRequestForm saveRequestForm = new SaveRequestForm();
                saveRequestForm.setIndex(ElasticConstants.WEATHER_HOUR_DATA);
                saveRequestForm.setDocumentId(docId);
                saveRequestForm.setDocument(jsonObject);

                if (existsResult.getData()) {
                    elasticService.updateDocument(saveRequestForm);
                } else {
                    elasticService.indexDocument(saveRequestForm);
                }
            }
        }
    }

    /**
     * 保存未来7天天气数据
     * @param weatherRegionCode 区域code
     * @param future7dData 天气数据
     */
    private void saveFuture7dData(WeatherRegionCode weatherRegionCode, List<CountyWeatherDto.FutureDayWeather> future7dData) {
        if (CollectionUtils.isEmpty(future7dData)) {
            return;
        }

        for (CountyWeatherDto.FutureDayWeather data : future7dData) {
            String docId = weatherRegionCode.getProvince() + ":" + weatherRegionCode.getCity() + ":"
                    + weatherRegionCode.getCounty() + ":" + data.getTime().substring(0, 10);

            JSONObject jsonObject = JSONObject.from(data);
            jsonObject.put("time", data.getTime().substring(0, 10));
            jsonObject.put("province", weatherRegionCode.getProvince());
            jsonObject.put("city", weatherRegionCode.getCity());
            jsonObject.put("county", weatherRegionCode.getCounty());
            jsonObject.put("updateTime", new Date().getTime());

            HttpResult<Boolean> existsResult = elasticService.documentExists(ElasticConstants.WEATHER_DAY_DATA, docId);
            if (Objects.nonNull(existsResult) && existsResult.getCode() == ResponseCode.SUCCESS.getCode() && Objects.nonNull(existsResult.getData())) {
                SaveRequestForm saveRequestForm = new SaveRequestForm();
                saveRequestForm.setIndex(ElasticConstants.WEATHER_DAY_DATA);
                saveRequestForm.setDocumentId(docId);
                saveRequestForm.setDocument(jsonObject);

                if (existsResult.getData()) {
                    elasticService.updateDocument(saveRequestForm);
                } else {
                    elasticService.indexDocument(saveRequestForm);
                }
            }
        }


    }
}
